package com.bsj.travel.mqtt.consumer.rocketmq;

import cn.hutool.core.net.NetUtil;
import cn.hutool.core.text.CharPool;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson2.JSON;
import com.bsj.travel.config.rocketmq.constant.ConsumerGroupConstant;
import com.bsj.travel.config.rocketmq.producer.BaseMqPushUtil;
import com.bsj.travel.constant.BaseServerTopicConstants;
import com.bsj.travel.def.common.DTO.mqtt.MqttEventPublish;
import com.bsj.travel.def.monitor.CloudLog;
import com.bsj.travel.def.mqtt.message.SendMqttMsg;
import com.bsj.travel.mqtt.enums.RedisKeys;
import com.bsj.travel.util.ip.GetEth0IpUtil;
import lombok.extern.slf4j.Slf4j;
import net.dreamlu.iot.mqtt.codec.MqttQoS;
import net.dreamlu.iot.mqtt.core.server.MqttServer;
import net.dreamlu.iot.mqtt.core.server.MqttServerCreator;
import net.dreamlu.mica.core.utils.INetUtil;
import net.dreamlu.mica.redis.cache.MicaRedisCache;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;

import java.util.List;

/**
 * @author : zhuojie.c
 * @version V1.0
 * @Description: MQTT服务端与设备交互服务端下行动作消息消费(集群消费)[消息下行][服务端 -> 设备]
 * @date Date : 2024/12/26 16:32
 */
@Slf4j
@Configuration
public class MqttDownActionMsgRocketmqConsumer {

    @Autowired
    private BaseMqPushUtil baseMqPushUtil;

    @Autowired
    private MicaRedisCache micaRedisCache;

    @Autowired
    private MqttServer mqttServer;

    @Autowired
    private MqttServerCreator serverCreator;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * mqtt设备交互服务端下行动作
     *
     * @return
     * @throws Exception
     */
    @Bean
    @DependsOn("rocketMQTemplate")
    public DefaultMQPushConsumer mqttDownActionMsgConsumer() throws Exception {
        // 节点
        String node = GetEth0IpUtil.getHostAddress();
        // 消费组
        String consumerGroup = ConsumerGroupConstant.TRAVEL_MQTT_DOWN_MSG_GROUP + "-" + node.replaceAll("\\.", "-");
        // 过滤标识
        String tag = BaseServerTopicConstants.TRAVEL_MQTT_MSG_DOWN_TAG + "_" + node;
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        // 设置NameServer地址
        consumer.setNamesrvAddr(rocketMQTemplate.getProducer().getNamesrvAddr());
        // 设置最大重试次数
        consumer.setMaxReconsumeTimes(3);
        // 订阅主题并设置选择器表达式
        consumer.subscribe(BaseServerTopicConstants.TRAVEL_MQTT_MSG_TOPIC, tag);
        // 设置集群模式
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 注册并发消息监听器
        MqttDownActionMsgListenerConcurrently concurrently = new MqttDownActionMsgListenerConcurrently();
        consumer.registerMessageListener(concurrently);
        // 启动消费者
        consumer.start();
        log.info("mqtt设备交互服务端下行MQ启动： consumerGroup：{},topic: {},tag: {}", consumerGroup, BaseServerTopicConstants.TRAVEL_MQTT_MSG_TOPIC, tag);
        return consumer;
    }

    /**
     * 消息消费
     */
    class MqttDownActionMsgListenerConcurrently implements MessageListenerConcurrently {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExt, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
            for (MessageExt ext : messageExt) {
                try {
                    String message = new String(ext.getBody());
                    log.info("TravelCloud-Mqtt物联网平台数据,下行动作消费: {}", message);
                    MqttEventPublish event = JSON.parseObject(message, MqttEventPublish.class);
                    String clientId = micaRedisCache.hGet(RedisKeys.CONNECT_STATUS.getKey(serverCreator.getIp()), event.getClient());
                    Object data = event.getData();
                    String payload;
                    if (data instanceof String) {
                        payload = data.toString();
                    } else {
                        payload = JSON.toJSONString(data);
                    }
                    SendMqttMsg sendMqttMsg = SendMqttMsg.builder()
                            .username(event.getClient())
                            .clientId(clientId)
                            .payload(payload)
                            .topic(event.getTopic())
                            .qos(event.getQos())
                            .timestamp(System.currentTimeMillis())
                            .build();
                    boolean publish = mqttServer.publish(clientId, sendMqttMsg.getTopic(),
                            sendMqttMsg.getPayload().getBytes(), MqttQoS.valueOf(sendMqttMsg.getQos()));
                    log.info("消息下行发送结果： {}", publish);
                    //消息下行结果写入时序数据
                    //转交给日志处理器
                    CloudLog cloudLog = new CloudLog();
                    cloudLog.setTs(System.currentTimeMillis());
                    //生成随机30位字符串当做事务ID
                    cloudLog.setTraceId(RandomUtil.randomString(30));
                    cloudLog.setServeType(1);
                    cloudLog.setOperate(sendMqttMsg.getTopic());
                    cloudLog.setIdentify(sendMqttMsg.getUsername());
                    cloudLog.setContent(sendMqttMsg.getPayload());
                    if (publish) {
                        //在线/下发成功
                        cloudLog.setCode(200);
                    } else {
                        //离线/下发失败
                        cloudLog.setCode(9201);
                    }
                    baseMqPushUtil.sendTopicBytes(BaseServerTopicConstants.TRAVEL_MQTT_LOG_MSG_RECORD_TOPIC,
                            BaseServerTopicConstants.TRAVEL_MQTT_LOG_MSG_RECORD_TAG, com.alibaba.fastjson2.JSON.toJSONBytes(cloudLog));
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; // 消费失败，稍后重试
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 消费成功
        }
    }
}
